package com.alison.datastream.chapter6_timeAndWaterMark;

import com.alison.datastream.domain.TaxiRide;
import com.alison.datastream.util.TaxiRideGenerator;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * @Author alison
 * @Date 2024/4/4 10:53
 * @Version 1.0
 * @Description
 */
public class TimeAndWaterMarkApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<TaxiRide> rides = env.addSource(new TaxiRideGenerator());
        // the WatermarkStrategy specifies how to extract timestamps and generate watermarks
        WatermarkStrategy<TaxiRide> taxiRideWatermarkStrategy = WatermarkStrategy.<TaxiRide>forBoundedOutOfOrderness(Duration.ofSeconds(60))
                .withTimestampAssigner(new SerializableTimestampAssigner<TaxiRide>() {
                    @Override
                    public long extractTimestamp(TaxiRide ride, long streamRecordTimestamp) {
                        return ride.getEventTimeMillis();
                    }
                });
        rides.assignTimestampsAndWatermarks(taxiRideWatermarkStrategy)
                .keyBy(ride -> ride.rideId)
                        .max(1).print();
        env.execute("watermarkstrategy");

    }
}
